package com.base.task;

import com.base.task.em.BatchTypeEm;
import com.base.task.entity.BatchResult;
import com.base.task.entity.BatchFilter;
import com.base.task.entity.BatchParam;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @ Author     ：山水之间-wxs
 * @ Description：批次服务抽象类
 * @ Date       ：Created in 9:57 2021/11/4
 * @ Modified By：wxs
 * @ Version    ：1.0
 */
@Data
@Slf4j
public abstract class AbstractBatchServer implements BatchServerFactory {

    protected BatchResult result;
    protected BatchFilter filter;
    protected BatchParam param;
    protected IBatch batchService;
    protected ThreadPoolExecutor threadPoolExecutor;

    public AbstractBatchServer(BatchParam batchParam, IBatch batchService) {
        this.param = batchParam;
        result = BatchResult.builder();
        result.setBatchParam(batchParam);
        filter = BatchFilter.builder();
        filter.setBatchParam(batchParam);
        filter.setSize();
        this.batchService = batchService;
    }

    @Override
    public int getTotalCount() {
        if (this.result.getTotalCount() == null) {
            this.result.setTotalCount(this.batchService.totalCount(filter));
        }
        if (this.result.getTotalCount() == null) {
            this.result.setTotalCount(0);
        }
        return this.result.getTotalCount();
    }

    protected abstract void buildExecutor();

    protected void buildThreadPool() {
        this.threadPoolExecutor = new ThreadPoolExecutor(this.param.getThreadMaxNum(), this.param.getThreadMaxNum(), 0,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
    }

    protected void closeThreadPool() {
        if (this.threadPoolExecutor != null && !this.threadPoolExecutor.isShutdown()) {
            this.threadPoolExecutor.shutdown();
        }
    }

    @Override
    public BatchResult run(BatchTypeEm batchType) {
        this.result.setBgnTime(LocalDateTime.now());
        if (getTotalCount() <= 0) {
            this.result.setEndTime(LocalDateTime.now());
            return this.result;
        }
        try {
            this.buildExecutor();
            int runCount = 0; int pageNum = 0;
            List<CompletableFuture<Integer>> results = new ArrayList<>();
            while (runCount < getTotalCount()) {
                if (batchType == BatchTypeEm.NORMAL) {
                    filter.setPageNo(pageNum); pageNum++;
                }
                List data = this.batchService.getData(filter);
                if (data == null || data.isEmpty()) {  break; }
                runCount += data.size();
                for (int i = 0; i < this.result.getThreadMaxNum(); i++) {
                    int fisrtIndex = i * this.result.getQueryMaxNum();
                    if (fisrtIndex >= data.size()) {  break; }
                    int lastIndex = (data.size() < (i + 1) * this.result.getQueryMaxNum())? data.size() : (i + 1) * this.result.getQueryMaxNum();
                    List handleData = data.subList(fisrtIndex, lastIndex);
                    results.add(CompletableFuture.supplyAsync(() -> this.batchService.handle(handleData), this.threadPoolExecutor).exceptionally(e-> {
                        log.error("[" + batchType.name() + "-BATCH]: handle BatchException,param:{}", this.getParam().toString(), e);
                        return 0;
                    }));
                }
                this.result.setSuccessCount(results.stream().map(CompletableFuture::join).reduce(Integer::sum).orElse(0));
                results.clear();
            }
            this.result.setEndTime(LocalDateTime.now());
            Duration duration = Duration.between(this.result.getBgnTime(), this.result.getEndTime());
            this.result.setTotalRunTime(duration.getSeconds());
        } finally {
            this.closeThreadPool();
        }
        return this.result;
    }

}
